package edu.gatech.PI2Hbase;


import java.io.FileOutputStream;
import java.io.IOException;
import java.io.PrintWriter;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;

public class PiMapreduce {
	


	public static ImmutableBytesWritable tagFilter(ImmutableBytesWritable tagname){
		

		String result = Bytes.toString(tagname.get());

		if(result.contains("PhaseXY_OutletVoltage")&& result.endsWith("1"))
			 result=result.substring(0, 3)+ "BC6-OutletVoltage";
		
		if(result.contains("PhaseXY_OutletVoltage")&& result.endsWith("8"))
			 result=result.substring(0, 3)+ "BC5-OutletVoltage";
		
		if(result.contains("PhaseXZ_OutletVoltage")&& result.endsWith("1"))
			 result=result.substring(0, 3)+ "BC4-OutletVoltage";
		
		if(result.contains("PhaseXZ_OutletVoltage")&& result.endsWith("8"))
			 result=result.substring(0, 3)+ "BC3-OutletVoltage";
		
		if(result.contains("PhaseYZ_OutletVoltage")&& result.endsWith("1"))
			 result=result.substring(0, 3)+ "BC2-OutletVoltage";
		
		if(result.contains("PhaseYZ_OutletVoltage")&& result.endsWith("8"))
			 result=result.substring(0, 3)+ "BC1-OutletVoltage";
		
		if(result.contains("PhaseXY_OutletCurrent")&& result.endsWith("1"))
			 result=result.substring(0, 3)+ "BC6-OutletCurrent";
		
		if(result.contains("PhaseXY_OutletCurrent")&& result.endsWith("8"))
			 result=result.substring(0, 3)+ "BC5-OutletCurrent";
		
		if(result.contains("PhaseXZ_OutletCurrent")&& result.endsWith("1"))
			 result=result.substring(0, 3)+ "BC4-OutletCurrent";
		
		if(result.contains("PhaseXZ_OutletCurrent")&& result.endsWith("8"))
			 result=result.substring(0, 3)+ "BC3-OutletCurrent";
		
		if(result.contains("PhaseYZ_OutletCurrent")&& result.endsWith("1"))
			 result=result.substring(0, 3)+ "BC2-OutletCurrent";
		
		if(result.contains("PhaseYZ_OutletCurrent")&& result.endsWith("8"))
			 result=result.substring(0, 3)+ "BC1-OutletCurrent";
		
		if(result.contains("PhaseXY_OutletEnergy")&& result.endsWith("1"))
			 result=result.substring(0, 3)+ "BC6-OutletEnergy";
		
		if(result.contains("PhaseXY_OutletEnergy")&& result.endsWith("8"))
			 result=result.substring(0, 3)+ "BC5-OutletEnergy";
		
		if(result.contains("PhaseXZ_OutletEnergy")&& result.endsWith("1"))
			 result=result.substring(0, 3)+ "BC4-OutletEnergy";
		
		if(result.contains("PhaseXZ_OutletEnergy")&& result.endsWith("8"))
			 result=result.substring(0, 3)+ "BC3-OutletEnergy";
		
		if(result.contains("PhaseYZ_OutletEnergy")&& result.endsWith("1"))
			 result=result.substring(0, 3)+ "BC2-OutletEnergy";
		
		if(result.contains("PhaseYZ_OutletEnergy")&& result.endsWith("8"))
			 result=result.substring(0, 3)+ "BC1-OutletEnergy";
		
		if(result.contains("PhaseXY_OutletPower")&& result.endsWith("1"))
			 result=result.substring(0, 3)+ "BC6-OutletPower";
		
		if(result.contains("PhaseXY_OutletPower")&& result.endsWith("8"))
			 result=result.substring(0, 3)+ "BC5-OutletPower";
		
		if(result.contains("PhaseXZ_OutletPower")&& result.endsWith("1"))
			 result=result.substring(0, 3)+ "BC4-OutletPower";
		
		if(result.contains("PhaseXZ_OutletPower")&& result.endsWith("8"))
			 result=result.substring(0, 3)+ "BC3-OutletPower";
		
		if(result.contains("PhaseYZ_OutletPower")&& result.endsWith("1"))
			 result=result.substring(0, 3)+ "BC2-OutletPower";
		
		if(result.contains("PhaseYZ_OutletPower")&& result.endsWith("8"))
			 result=result.substring(0, 3)+ "BC1-OutletPower";
		
		ImmutableBytesWritable tag = new ImmutableBytesWritable(result.getBytes(),0,result.length());
		
		return tag;
	}
	


	static class Mapper extends TableMapper<ImmutableBytesWritable,Put>{
		
        private int numRecords = 0;
        double value;
		String tag;
        @Override
        public void map(ImmutableBytesWritable tagname, Result values, Context context) throws IOException {
            // extract userKey from the compositeKey (userId + counter)
    	     FileOutputStream foutstream = new FileOutputStream("/tmp/reducelatency");
    		 PrintWriter fout = new PrintWriter(foutstream,true);

    		String tagvalue = Bytes.toString(values.getValue(Bytes.toBytes("Info"), Bytes.toBytes("Value")));
    		if(tagvalue.contains("Not")||tagvalue.contains("null"))
    			value = 0L;
    		else
    			value = Double.parseDouble(tagvalue); 
    		
    		
    		long timestamp = values.list().get(0).getTimestamp();
    		
    		fout.println(timestamp);
    		fout.println(System.currentTimeMillis()-timestamp);
    		Put intervalue = new Put(tagname.get());
    		
    		intervalue.add(Bytes.toBytes("Info"), Bytes.toBytes("Value"), timestamp,Bytes.toBytes(value));
        	
        	try {
                context.write(tagFilter(tagname), intervalue);
               
            } catch (InterruptedException e) {
                throw new IOException(e);
            }
            numRecords++;
            if ((numRecords % 10000) == 0) {
                context.setStatus("mapper processed " + numRecords + " records so far");
            }
        }
		
	}
	
	static class Reducer extends TableReducer<ImmutableBytesWritable,Put,ImmutableBytesWritable>{
		       
			   public void reduce(ImmutableBytesWritable key, Iterable<Put> values, Context context)
		        throws IOException, InterruptedException {
				    double sum = 0;
				    long timestamp = 0L;
				    for (Put val : values) {
				        sum +=Bytes.toDouble(val.get(Bytes.toBytes("Info"), Bytes.toBytes("Value")).get(0).getValue());
				        if(timestamp < val.getTimeStamp())
				        	timestamp = val.getTimeStamp();
				    }
		
			    Put put = new Put(key.get());
			    
			    put.add(Bytes.toBytes("Info"), Bytes.toBytes("Value"), timestamp,Bytes.toBytes(sum));
			    System.out.println("tagname:"+ Bytes.toString(key.get())+"  value: "+ sum);
			    context.write(key, put);
		}
		
		
	}
}
